package com.alan.mr.weibo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.text.NumberFormat;
import java.util.HashMap;
import java.util.Map;

/**
 * Created by Alan on 2017/10/2.
 */
public class ThirdMapper extends Mapper<LongWritable, Text, Text, Text> {

    //存放微博总数, 将小数据缓存进内存, 预加载
    public static Map<String, Integer> cmap = null;
    //存放df
    public static Map<String, Integer> df = null;

    // 在初始化类时执行, 将数据预加载进map
    protected void setup(Context context)
            throws IOException, InterruptedException {
        System.out.println("*****");
        if (cmap == null || cmap.size() == 0 || df == null || df.size() == 0) {
            URI[] cacheFiles = context.getCacheFiles();
            if (cacheFiles != null) {
                for (URI uri : cacheFiles) {
                    if (uri.getPath().endsWith("part-r-00003")) {
                        Path path = new Path(uri.getPath());
                        // 获取文件
                        Configuration configuration = context.getConfiguration();
                        FileSystem fs = FileSystem.get(configuration);
                        FSDataInputStream open = fs.open(path);
                        BufferedReader reader = new BufferedReader(new InputStreamReader(open));
//                            BufferedReader reader = new BufferedReader(new FileReader(path.getName()));
                        String line = reader.readLine();
                        if (line.startsWith("count")) {
                            String[] split = line.split("\t");
                            cmap = new HashMap<>();
                            cmap.put(split[0], Integer.parseInt(split[1].trim()));
                        }
                        reader.close();
                    }else if (uri.getPath().endsWith("part-r-00000")) {
                        df = new HashMap<>();
                        Path path = new Path(uri.getPath());

                        // 获取文件
                        Configuration configuration = context.getConfiguration();
                        FileSystem fs = FileSystem.get(configuration);
                        FSDataInputStream open = fs.open(path);
                        BufferedReader reader = new BufferedReader(new InputStreamReader(open));
//                            BufferedReader reader = new BufferedReader(new FileReader(path.getName()));

                        String line = null;
                        while ((line = reader.readLine()) != null) {
                            String[] ls = line.split("\t");
                            df.put(ls[0], Integer.parseInt(ls[1].trim()));
                        }
                        reader.close();
                    }
                }
            }
        }
    }

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
            throws IOException, InterruptedException {
        // 获取分片
        FileSplit inputSplit = (FileSplit) context.getInputSplit();

        if (!inputSplit.getPath().getName().contains("part-r-00003")) {
            String[] values = value.toString().trim().split("\t");

            if (values.length >= 2) {

                int tf = Integer.parseInt(values[1].trim());

                String[] ss = values[0].split("_");
                if (ss.length >= 2) {
                    String word = ss[0];
                    String id = ss[1];

                    // 公式
                    Double s = tf * Math.log(cmap.get("count")) / df.get(word);
                    NumberFormat format = NumberFormat.getInstance();
                    // 取小数点后5位
                    format.setMaximumFractionDigits(5);

                    context.write(new Text(id), new Text(word + ": " + format.format(s)));
                }else {
                    System.out.println(value.toString() + "------");
                }
            }
        }
    }
}
